﻿/* This is a customized version of SizeControlledConnectionPool.cs from the Aquiles project.
 * Aquiles can be found at http://aquiles.codeplex.com/ */

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Lucandra.Net.Connection.Endpoint;
using System.Threading;
using System.Diagnostics;
using System.Net.Sockets;
using Lucandra.Configuration;
using System.Configuration;
using Lucandra.Net;

namespace Lucandra.Net.Connection.Pooling
{
    public static class CassandraConnectionPool
    {
        private const int DefaultDueTime = 20000;
        private const int DefaultPeriodicTime = 5000;
        private const int DefaultMinConnections = 10;
        private const int DefaultMaxConnections = 1000;
        private const int DefaultMagicNumber = 7;
        private const int DefaultMaxClientPollAttempts = 0;

        private static Queue<CassandraClient> idleClients;
        private static HashSet<CassandraClient> referencedClients;
        private static IEndpointManager endpointManager;
        private static Timer controlIdleClientSizeTimer;

        internal static int minConnections;
        internal static int maxConnections;
        private static int magicNumber;
        private static int maxClientPollAttempts;
        internal static bool isInitialized = false;

        public static double TotalClientsCreated { get; private set; }
        public static double TotalClientsDestroyed { get; private set; }

        public static int ManagedClientQuantity
        {
            get { return idleClients.Count + referencedClients.Count; }
        }

        public static int BorrowedConnectionCount
        {
            get { return referencedClients.Count; }
        }

        public static int IdleConnectionCount
        {
            get { return idleClients.Count; }
        }

        static CassandraConnectionPool()
        {
            endpointManager = new RoundRobinEndpointManager();

            referencedClients = new HashSet<CassandraClient>();
            idleClients = new Queue<CassandraClient>();
        }

        #region Methods [public]

        public static void Initialize()
        {
            if (isInitialized)
                return;

            try
            {
                var config = LucandraConfigurationSection.GetConfig();

                minConnections = config.ConnectionPool.MinConnections;
                maxConnections = config.ConnectionPool.MaxConnections;
                magicNumber = config.ConnectionPool.MagicNumber;
                maxClientPollAttempts = config.ConnectionPool.MaxClientPollAttempts;

                if (config.ConnectionPool.CassandraEndpoints.Count == 0)
                    throw new ConfigurationErrorsException("There are no cassandra endpoints configured in the config file.");

                foreach (CassandraEndpointElement endpoint in config.ConnectionPool.CassandraEndpoints)
                    endpointManager.AddEndpoint(new CassandraEndpoint(endpoint.Host, endpoint.Port, endpoint.Timeout));

                controlIdleClientSizeTimer = new Timer(new TimerCallback(ControlIdleClientSizeTimerCallback), null, DefaultDueTime, Timeout.Infinite);
                isInitialized = true;
            }
            catch (ConfigurationErrorsException ex)
            {
                Debug.WriteLine("[Cassandra Pool] The Lucandra configuration section is invalid: " + ex.Message);
                throw;
            }

            CreateMinConnections();
        }

        #endregion
        #region Methods [private]

        private static void CreateMinConnections()
        {
            // Create the minimum number of clients.
            lock (idleClients)
            {
                for (int i = ManagedClientQuantity; i < minConnections; i++)
                {
                    var client = CreateNewClient();
                    if (client != null)
                        idleClients.Enqueue(client);
                    else
                        throw new LucandraException("Couldn't create a connection for the Lucandra.Net.Connection pool.");
                }
            }
        }

        private static void AssertInitialized()
        {
            if (!isInitialized)
                Initialize();
        }

        private static void Destroy(CassandraClient cassandraClient)
        {
            Debug.WriteLine("[Cassandra Pool] Destroying {0}.", cassandraClient);
            if (cassandraClient.IsOpened)
                cassandraClient.CloseTransport();
            TotalClientsDestroyed++;
        }

        private static void MarkAsReferenced(CassandraClient borrowedClient)
        {
            lock (referencedClients)
                referencedClients.Add(borrowedClient);
        }

        private static void MarkAsIdle(CassandraClient client)
        {
            lock (idleClients)
            {
                if (!idleClients.Contains(client) && client.IsOpened)
                    idleClients.Enqueue(client);
            }
        }

        private static void UnmarkAsReferenced(CassandraClient client)
        {
            lock (referencedClients)
            {
                if (referencedClients.Contains(client))
                    referencedClients.Remove(client);
            }
        }

        private static CassandraClient PollClient()
        {
            CassandraClient borrowedClient = null;
            lock (idleClients)
                if (idleClients.Count > 0)
                    borrowedClient = idleClients.Dequeue();
            return borrowedClient;
        }

        private static CassandraClient CreateNewClient()
        {
            int attempts = 0;

            CassandraClient borrowedClient = null;
            CassandraEndpoint endpoint = null;

            do
            {
                endpoint = endpointManager.Retrieve();
                if (endpoint != null)
                {
                    Debug.WriteLine("[Cassandra Pool] Creating client for endpoint {0}.", endpoint);
                    borrowedClient = CassandraConnectionFactory.Create(endpoint);

                    try
                    {
                        borrowedClient.OpenTransport();
                    }
                    catch (SocketException)
                    {
                        endpointManager.Ban(endpoint);
                        borrowedClient = null;
                    }
                }
                else
                    Debug.WriteLine("[Cassandra Pool] No endpoints available.  Cluster may be down or no endpoints defined.");

                attempts++;
            }
            while (endpoint != null && borrowedClient == null && attempts < 10);

            if (borrowedClient == null)
                throw new LucandraException("[Cassandra Pool] Could not create new client after 10 attempts.");
            else
                TotalClientsCreated++;

            return borrowedClient;
        }

        private static void StopTimer()
        {
            controlIdleClientSizeTimer.Change(Timeout.Infinite, Timeout.Infinite);
        }

        private static void StartTimer(int dueTime)
        {
            controlIdleClientSizeTimer.Change(dueTime, Timeout.Infinite);
        }

        #endregion
        #region Methods [internal]

        internal static CassandraClient Borrow()
        {
            AssertInitialized();

            CassandraClient borrowedClient = null;
            int retryCount = 0;

            do
            {
                Debug.WriteLine("[Cassandra Pool] Status: Total={0}, Free={1}, Maximum={2}.", referencedClients.Count, idleClients.Count, maxConnections);
                borrowedClient = PollClient();

                if (borrowedClient == null)
                {
                    if (ManagedClientQuantity < maxConnections)
                    {
                        Debug.WriteLine("[Cassandra Pool] No client could be borrowed.  Retrieving a new endpoint.");
                        borrowedClient = CreateNewClient();
                    }
                    else
                        throw new LucandraException(string.Format("[Cassandra Pool] Pool size is at maximum capacity ({0}/{1}).", ManagedClientQuantity, maxConnections));
                }
                else if (!borrowedClient.IsOpened)
                {
                    Debug.WriteLine("[Cassandra Pool] Destroying client {0} because it is not open.", borrowedClient);
                    Destroy(borrowedClient);
                    borrowedClient = null;
                }

                if (borrowedClient != null)
                {
                    Debug.WriteLine("[Cassandra Pool] Marking client {0} as referenced.", borrowedClient);
                    MarkAsReferenced(borrowedClient);
                    Debug.WriteLine("[Cassandra Pool] Opening {0} for use.", borrowedClient);
                }

                retryCount++;
            }
            while (retryCount < maxClientPollAttempts && borrowedClient == null);

            if (borrowedClient == null)
                throw new LucandraException("[Cassandra Pool] Could not retrieve a connection from the pool.");

            borrowedClient.IsBorrowed = true;
            return borrowedClient;
        }

        internal static void Release(CassandraClient cassandraClient)
        {
            AssertInitialized();

            Debug.WriteLine("[Cassandra Pool] Unmarking client {0} as referenced.", cassandraClient);
            UnmarkAsReferenced(cassandraClient);
            Debug.WriteLine("[Cassandra Pool] Marking client {0} as idle.", cassandraClient);
            MarkAsIdle(cassandraClient);

            cassandraClient.IsBorrowed = false;
        }

        internal static void Invalidate(CassandraClient cassandraClient)
        {
            AssertInitialized();

            Debug.WriteLine("[Cassandra Pool] Unmarking client {0} as referenced.", cassandraClient);
            UnmarkAsReferenced(cassandraClient);
            Debug.WriteLine("[Cassandra Pool] Marking endpoint {0} as invalid.", cassandraClient.Endpoint);
            endpointManager.Ban(cassandraClient.Endpoint);
            Destroy(cassandraClient);

            cassandraClient.IsInvalidated = true;
        }

        #endregion

        #region Timer Callbacks

        private static void ControlIdleClientSizeTimerCallback(object state)
        {
            bool completedGracefully = false;
            Debug.WriteLine("[Cassandra Pool] Stopping ControlIdleClientSize timer.");

            //int difference = (maxConnections - minConnections) / (magicNumber > 0 ? magicNumber : DefaultMagicNumber);
            int toDestroy = (int)Math.Floor((ManagedClientQuantity - minConnections) * 0.25);
            int toRefresh = (int)Math.Floor(ManagedClientQuantity * 0.05);

            StopTimer();

            var clientsToDestroy = new List<CassandraClient>();

            try
            {
                if (idleClients.Count > minConnections)
                {
                    Debug.WriteLine("[Cassandra Pool] Performing cleanup.");
                    lock (idleClients)
                    {
                        for (int i = 0; i < toDestroy && idleClients.Count > minConnections; i++)
                            clientsToDestroy.Add(idleClients.Dequeue());
                    }
                    Debug.WriteLine("[Cassandra Pool] Destroying {0} clients.", clientsToDestroy.Count);
                    foreach (var client in clientsToDestroy)
                        Destroy(client);
                }

                clientsToDestroy.Clear();
                lock (idleClients)
                {
                    for (int i = 0; i < toRefresh; i++)
                        clientsToDestroy.Add(idleClients.Dequeue());
                }

                Debug.WriteLine("[Cassandra Pool] Performing refresh of 5% of client pool ({0}).", clientsToDestroy.Count);
                foreach (var client in clientsToDestroy)
                    Destroy(client);

                Debug.WriteLine("[Cassandra Pool] Creating minimum connections.");
                CreateMinConnections();
                    
                completedGracefully = true;
            }
            catch (Exception ex)
            {
                Debug.WriteLine(ex);
            }
            finally
            {
                if (completedGracefully)
                    Debug.WriteLine("[Cassandra Pool] Cleanup completed.");
                else
                    Debug.WriteLine("[Cassandra Pool] Cleanup failed.");

                StartTimer(DefaultPeriodicTime);
            }
        }

        #endregion
    }
}
